package com.atguigu.sql.connector;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Flink05_JdbcConnector {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(2000L);

        //流表环境
        StreamTableEnvironment streamTableEnv = StreamTableEnvironment.create(env);

        //JdbcSource
        String sourceTable =
                "create table t_source ( " +
                        "id STRING," +
                        "vc INT ," +
                        "ts BIGINT" +
                ") WITH (" +
                        "'connector' = 'jdbc'," +
                        "'driver' = 'com.mysql.cj.jdbc.Driver', " +
                        "'url' = 'jdbc:mysql://hadoop102:3306/test'," +
                        "'username'= 'root' ," +
                        "'password' ='000000'," +
                        "'table-name'= 't_ws'" +
                ")"; //从表t_ws中读数据放到t_source中

        streamTableEnv.executeSql(sourceTable);

        //查询数据
        Table resultTable = streamTableEnv.sqlQuery("select id , vc, ts from t_source");
        //todo 打印查询到的数据
        //resultTable.execute().print();
        //将查询的结果暂时存放在中间视图中
        streamTableEnv.createTemporaryView("result_table",resultTable);


        //将查询到的表的结果写出到Mysql
        //String sinkTable =
        //        "create table t_sink_ws ( " +
        //                " id STRING," +
        //                " vc INT , " +
        //                " ts BIGINT , " +
        //                " PRIMARY KEY (id) NOT ENFORCED " +
        //        " ) WITH ( " +
        //                " 'connector' = 'jdbc', " +
        //                " 'driver' = 'com.mysql.cj.jdbc.Driver' , " +
        //                " 'url' = 'jdbc:mysql://hadoop102:3306/test' ,  " +
        //                " 'username' = 'root' , " +
        //                " 'password' = '000000' , " +
        //                " 'table-name' = 't_sink_ws' " +
        //        " ) " ;

        //jdbc连接器连接mysql要写出的表，将表的结果写出到Mysql,使用之前先定义
        String sinkTable =
                "create table t_sink_ws (" +
                        "id STRING," +
                        "sum_vc INT," +
                        "PRIMARY KEY (id) NOT ENFORCED" +
                ") WITH (" +
                        "'connector' = 'jdbc'," +
                        "'driver' = 'com.mysql.cj.jdbc.Driver', " +
                        "'url' = 'jdbc:mysql://hadoop102:3306/test'," +
                        "'username'= 'root' ," +
                        "'password' ='000000'," +
                        "'table-name'= 't_sink_ws2'" +
               ")"; //将数据写到t_sink_ws2

        streamTableEnv.executeSql(sinkTable);

        //写出
        streamTableEnv.executeSql("insert into t_sink_ws select id ,sum(vc)  from result_table group by id" );

    }
}
